package com.go.udfs;

import com.alibaba.fastjson.JSON;
import com.go.pojo.DataReading;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * ClassName: JsonToPojoFromKafka
 * Description:
 * Date: 2022/1/8
 * @author: Cason
 */
public class JsonToPojoFromKafka extends ProcessFunction<Tuple2<String, String>, DataReading> {

    @Override
    public void processElement(Tuple2<String, String> value, Context ctx, Collector<DataReading> out) throws Exception {
        try {
            System.out.println("stap 1");
            System.out.println(value.f1);
            DataReading dataBean = JSON.parseObject(value.f1, DataReading.class);
            dataBean.setUid(value.f0);

            out.collect(dataBean);

        } catch (Exception e) {
            System.out.println("got something wrong");
        }
    }
}
